package org.alis.smallcc.alis.tools.lock.distributed;

import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

import org.alis.smallcc.alis.tools.lock.exception.LockException;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.zookeeper.CreateMode;


import lombok.extern.slf4j.Slf4j;

@Slf4j
public class ZookeeperLock implements Lock {
	private final CuratorFramework client;

	private final InterProcessMutex mutex;

	private final String path;

	private final ThreadPoolExecutor executor;

	public ZookeeperLock(CuratorFramework client, String path,ThreadPoolExecutor executor) {
		this(client,path,new InterProcessMutex(client, path),executor);
	}
	public ZookeeperLock(CuratorFramework client, String path,InterProcessMutex mutex,ThreadPoolExecutor executor) {
		this.client = client;
		this.path = path;
		this.mutex = mutex;
		this.executor = executor;
	}

	@Override
	public void lock() {
		try {
			this.mutex.acquire();
		} catch (Exception e) {
			throw new LockException("枷锁锁异常", e);
		}
	}

	@Override
	public void lockInterruptibly() throws InterruptedException {
		boolean locked = false;
		while (!locked) {
			locked = tryLock();
		}
	}

	@Override
	public boolean tryLock() {
		try {
			return tryLock(1, TimeUnit.SECONDS);
		} catch (InterruptedException e) {
			log.error("线程异常中断");
			Thread.currentThread().interrupt();
		}
		return false;
	}

	@Override
	public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
		Future<String> future = null;
		try {
			long startTime = System.currentTimeMillis();

			future = this.executor.submit(() -> client.create().creatingParentContainersIfNeeded().withProtection()
					.withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(path));

			long waitTime = unit.toMillis(time);

			String ourPath = future.get(waitTime, TimeUnit.MILLISECONDS);

			if (ourPath == null) {
				future.cancel(true);
				return false;
			} else {
				waitTime = waitTime - (System.currentTimeMillis() - startTime);
				return this.mutex.acquire(waitTime, TimeUnit.MILLISECONDS);
			}
		} catch (TimeoutException e) {
			future.cancel(true);
			return false;
		} catch (Exception e) {
			throw new LockException("Failed to acquire mutex at  " + this.path, e);
		}
	}

	@Override
	public void unlock() {
		try {
			this.mutex.release();
		} catch (Exception e) {
			throw new LockException("Failed to release mutex at " + this.path, e);
		}
	}
	public boolean isAcquiredInThisProcess() {
        return this.mutex.isAcquiredInThisProcess();
    }
	@Override
	public Condition newCondition() {
		throw new UnsupportedOperationException("分布式锁基本版不给condition");
	}

}
